package rpc

import (
	"fmt"
	"log"
	"time"

	myactor "gitee.com/nggs/actor"
	p "gitee.com/nggs/network/protocol"
	"gitee.com/nggs/protoactor-go/actor"
	proto "github.com/gogo/protobuf/proto"
)

type MessageID = uint32

type IMessage interface {
	proto.Message
	Size() int
	Marshal() (dAtA []byte, err error)
	Unmarshal([]byte) error
	RPC()
	MessageID() MessageID
	MessageName() string
	ResetEx()
	JsonString() string
	ResponseMessageID() MessageID
	NeedMarkDirty() bool
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
type messageProducer func() IMessage
type messageRecycler func(IMessage)

type messageFactory struct {
	id       MessageID
	producer messageProducer
	recycler messageRecycler
}

func newMessageFactory(id MessageID, producer messageProducer, recycler messageRecycler) *messageFactory {
	return &messageFactory{
		id:       id,
		producer: producer,
		recycler: recycler,
	}
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
type IMessageFactoryManager interface {
	Register(iMsg IMessage, producer messageProducer, recycler messageRecycler)
	Produce(id MessageID) (IMessage, error)
	Recycle(iMsg IMessage) error
}

type messageFactoryManager struct {
	factories map[MessageID]*messageFactory
}

func newMessageFactoryManager() IMessageFactoryManager {
	m := &messageFactoryManager{
		factories: make(map[MessageID]*messageFactory),
	}
	return m
}

func (m *messageFactoryManager) Register(iMsg IMessage, producer messageProducer, recycler messageRecycler) {
	if producer == nil {
		log.Panicf("register message factory fail, producer == nil, id=[%v]", iMsg.MessageID())
	}

	if recycler == nil {
		log.Panicf("register message factory fail, recycler == nil, id=[%v]", iMsg.MessageID())
	}

	if f, ok := m.factories[iMsg.MessageID()]; ok {
		log.Panicf("duplicate message factory, id=[%v], factory=[%+v]", iMsg.MessageID(), f)
	}

	m.factories[iMsg.MessageID()] = newMessageFactory(iMsg.MessageID(), producer, recycler)
}

func (m *messageFactoryManager) Produce(id MessageID) (IMessage, error) {
	if factory := m.factories[id]; factory != nil {
		return factory.producer(), nil
	}
	return nil, fmt.Errorf("unsupported message, id=[%v]", id)
}

func (m *messageFactoryManager) Recycle(iMsg IMessage) error {
	if factory := m.factories[iMsg.MessageID()]; factory != nil {
		factory.recycler(iMsg)
		return nil
	}
	return fmt.Errorf("unsupported message, id=[%v]", iMsg.MessageID())
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
type MessageHandler func(sender *actor.PID, iRecv IMessage, ctx actor.Context, args ...interface{})

type IMessageDispatcher interface {
	IPromiseManager
	MustRegister(id MessageID, handler MessageHandler)
	Dispatch(iProtocol IProtocol, sender *actor.PID, iRecv IMessage, ctx actor.Context, args ...interface{}) error
}

////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
type messageDispatcher struct {
	dispatchers map[MessageID]MessageHandler

	promiseSeqGenerator PromiseSeq
	promiseMap          map[PromiseSeq]*Promise
}

func NewMessageDispatcher() IMessageDispatcher {
	return &messageDispatcher{
		dispatchers: map[MessageID]MessageHandler{},
		promiseMap:  map[PromiseSeq]*Promise{},
	}
}

func (m *messageDispatcher) MustRegister(id MessageID, handler MessageHandler) {
	if _, ok := m.dispatchers[id]; ok {
		log.Panicf("%v already register", id)
	}

	m.dispatchers[id] = handler
}

func (m *messageDispatcher) Dispatch(iProtocol IProtocol, sender *actor.PID, iRecv IMessage, ctx actor.Context, args ...interface{}) (err error) {
	switch recv := iRecv.(type) {
	case *C2S_Promise:
		requestMessage, e := iProtocol.Produce(recv.ReqMsgID)
		if e != nil {
			err = &ErrPromiseProduceRequestMessageFail{
				Seq:              recv.Seq,
				RequestMessageID: recv.ReqMsgID,
				Err:              e,
			}
			return
		}
		// todo 这里无法回收对象，因为业务层会浅拷贝来自rpc包的对象，如果回收对象，会导致各种问题
		//defer iProtocol.Recycle(requestMessage)

		// 反序列化收到的请求包
		e = requestMessage.Unmarshal(recv.ReqMsgData)
		if e != nil {
			err = &ErrPromiseUnmarshalRequestMessageFail{
				Seq:              recv.Seq,
				RequestMessageID: recv.ReqMsgID,
				Err:              e,
			}
			return
		}

		responseMessage, e := iProtocol.Produce(requestMessage.ResponseMessageID())
		if e != nil {
			err = &ErrPromiseProduceResponseMessageFail{
				Seq:               recv.Seq,
				RequestMessageID:  recv.ReqMsgID,
				ResponseMessageID: requestMessage.ResponseMessageID(),
				Err:               e,
			}
			return
		}
		// todo 这里无法回收对象，因为业务层会浅拷贝来自rpc包的对象，如果回收对象，会导致各种问题
		//defer iProtocol.Recycle(responseMessage)

		handler, ok := m.dispatchers[recv.ReqMsgID]
		if !ok {
			err = &p.ErrNoDispatcher{MessageID: recv.ReqMsgID}
			return
		}

		args = append(args, responseMessage)

		handler(sender, requestMessage, ctx, args...)

		send := Get_S2C_Promise()
		send.Seq = recv.Seq
		send.RespMsgID = responseMessage.MessageID()
		send.RespMsgData, e = responseMessage.Marshal()
		if e != nil {
			err = &ErrPromiseMarshalResponseMessageFail{
				Seq:               recv.Seq,
				RequestMessageID:  requestMessage.MessageID(),
				ResponseMessageID: responseMessage.MessageID(),
				Err:               e,
			}
			return
		}
		myactor.RootContext.Send(sender, send)

	case *S2C_Promise:
		promise, ok := m.promiseMap[recv.Seq]
		if !ok {
			err = &ErrPromiseNotExist{Seq: recv.Seq}
			return
		}

		responseMessage, e := iProtocol.Produce(promise.IRequestMessage.ResponseMessageID())
		if e != nil {
			err = &ErrPromiseProduceResponseMessageFail{
				Seq:               promise.seq,
				RequestMessageID:  promise.IRequestMessage.MessageID(),
				ResponseMessageID: promise.IRequestMessage.ResponseMessageID(),
				Err:               e,
			}
			promise.onComplete(err)
			return
		}
		// todo 这里无法回收对象，因为业务层会浅拷贝来自rpc包的对象，如果回收对象，会导致各种问题
		//defer iProtocol.Recycle(responseMessage)

		// 反序列化收到的回复包
		e = responseMessage.Unmarshal(recv.RespMsgData)
		if e != nil {
			err = &ErrPromiseUnmarshalResponseMessageFail{
				Seq:               promise.seq,
				RequestMessageID:  promise.IRequestMessage.MessageID(),
				ResponseMessageID: promise.IRequestMessage.ResponseMessageID(),
				Err:               e,
			}
			promise.onComplete(err)
			return
		}

		if promise.onResponse(responseMessage) {
			delete(m.promiseMap, promise.seq)
		}

	default:
		handler, ok := m.dispatchers[recv.MessageID()]
		if !ok {
			err = &p.ErrNoDispatcher{MessageID: recv.MessageID()}
			return
		}
		handler(sender, recv, ctx, args...)
	}
	return
}

func (m *messageDispatcher) nextPromiseSeq() PromiseSeq {
	m.promiseSeqGenerator += 1
	return m.promiseSeqGenerator
}

func (m *messageDispatcher) ExecutePromise(promise *Promise) {
	promise.seq = m.nextPromiseSeq()
	m.promiseMap[promise.seq] = promise
	promise.Execute()
	return
}

func (m *messageDispatcher) CheckAndRemovePromiseTimer(timerID myactor.TimerID, timerTag myactor.TimerID) {
	var now = time.Now()
	var needRemoveSeqMap = map[PromiseSeq]struct{}{}

	for _, promise := range m.promiseMap {
		if promise.completed {
			needRemoveSeqMap[promise.seq] = struct{}{}
			continue
		}
		if !promise.timeout.Before(now) {
			continue
		}
		needRemoveSeqMap[promise.seq] = struct{}{}
		promise.onComplete(&ErrPromiseTimeout{
			Seq:              promise.seq,
			RequestMessageID: promise.IRequestMessage.MessageID(),
		})
	}
	for seq, _ := range needRemoveSeqMap {
		delete(m.promiseMap, seq)
	}
}

func (m messageDispatcher) PromiseNum() int {
	return len(m.promiseMap)
}
